package com.example.java.kafka.kafkaclients;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;

public abstract class MessageConsumerGroup {
    // 使用固定大小线程池，至少支持 3 个消费者线程
    private static final ExecutorService threadPool = Executors.newFixedThreadPool(3);

    public static void main(String[] args) {
        MessageConsumerGroup messageConsumer = new MessageConsumerGroup() {
            @Override
            public void consumeMsg(String consumerData, String consumerName) throws Exception {
                Thread.sleep(100); // 模拟处理时间，观察并行效果
                System.out.println("Thread: " + Thread.currentThread().getName() + " | consumerName: " + consumerName +
                        " | Consume time: " + new Date().getTime() + " | Consume message: " + consumerData);
            }
        };
        // 启动 3 个消费者实例，加入同一个消费者组
        for (int i = 0; i < 3; i++) {
            messageConsumer.startListen(messageConsumer.kafkaConsumerInit(), "Consumer-" + i);
        }
        // 主线程不退出（等待消费者持续运行）
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 消费者实例
     */
    public KafkaConsumer<String, String> kafkaConsumerInit() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "43.139.50.31:9092");
        // 同一个 group.id → 属于同一个消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 增加会话和心跳配置，避免意外再平衡（可选优化）
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");
        // 创建消费者实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("producer-api-topic"));
        return consumer;
    }

    /**
     * 消费者监听
     *
     * @param consumer
     * @param consumerName
     */
    private void startListen(KafkaConsumer<String, String> consumer, String consumerName) {
        threadPool.submit(() -> {
            System.out.println(consumerName + " started.");
            try {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                    if (records.isEmpty()) {
                        System.out.println(consumerName + " - No records found.");
                        continue;
                    }
                    for (ConsumerRecord<String, String> record : records) {
                        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
                        if (kafkaMessage.isPresent()) {
                            consumeMsg(kafkaMessage.get(), consumerName);
                        }
                    }
                }
            } catch (WakeupException e) {
                System.out.println(consumerName + " is shutting down...");
            } catch (Exception e) {
                System.out.println(e.getMessage());
            } finally {
                try {
                    consumer.close();
                } catch (Exception e) {
                    System.out.println(e.getMessage());
                }
            }
        });
    }

    /**
     * 核心消费逻辑，由子类实现
     *
     * @param consumerMessage
     * @param consumerName
     */
    public abstract void consumeMsg(String consumerMessage, String consumerName) throws Exception;
}